#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <string.h>
#include <unistd.h>
#include "common.h"
#include "schema.h"
#include "lib.h"
#define BLOCK  (1024*1024*50)

extern void tableScan(struct tableNode *, struct relation *, struct statistic *);
extern struct tableNode* hashJoin(struct joinNode *, struct statistic *);
int main(int argc, char ** argv){

//initialize the gpu device
	int * tmp;
	cudaMalloc((void **)&tmp, 4);
	cudaFree(tmp);

	struct timespec start,end;
	struct timespec startAll;
	struct statistic pp;
	double totalTime = 0;
	pp.dimTransfer = 0;
	pp.factTransfer = 0;
	pp.totalTransfer = 0;
	clock_gettime(CLOCK_REALTIME,&startAll);

	struct tableNode *supplierTable;
	struct tableNode *customerTable;
	struct tableNode *partTable;
	struct tableNode *ddateTable;
	int outFd;
	int outSize;
	char *outTable;

	supplierTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	supplierTable->totalAttr = 1;
	supplierTable->attrType = (int *) malloc(sizeof(int)*1);
	supplierTable->attrSize = (int *) malloc(sizeof(int)*1);
	supplierTable->content = (char **) malloc(sizeof(char *)*1);
	supplierTable->attrSize[0] = sizeof(int);
	supplierTable->attrType[0] = INT;
	outFd = open("SUPPLIER0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	supplierTable->tupleNum = outSize/sizeof(int);
	supplierTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(supplierTable->content[0],outTable,outSize);
	close(outFd);
	supplierTable->tupleSize = 0 + sizeof(int);
	supplierTable->dataPos = MEM;

	struct relation supplierRel;
	supplierRel.hasWhere = 1;
	supplierRel.whereAttrNum = 1;
	supplierRel.whereAttrType = (int *)malloc(sizeof(int)*1);
	supplierRel.whereAttrSize = (int *)malloc(sizeof(int)*1);
	supplierRel.content = (char **)malloc(sizeof(char *)*1);
	supplierRel.whereAttrType[0] = STRING;
	supplierRel.whereAttrSize[0] = 12;
	outFd = open("SUPPLIER5",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	supplierRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(supplierRel.content[0],outTable,outSize);
	close(outFd);
	supplierRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(supplierRel.filter)->nested = 0;
	(supplierRel.filter)->andOr = EXP;
	(supplierRel.filter)->exp[0].index = 0;
	(supplierRel.filter)->exp[0].relation = EQ;
	strcpy((supplierRel.filter)->exp[0].content,"AMERICA");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan(supplierTable, &supplierRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);
	totalTime += (end.tv_sec - start.tv_sec) * BILLION + end.tv_nsec - start.tv_nsec;

	customerTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	customerTable->totalAttr = 2;
	customerTable->attrType = (int *) malloc(sizeof(int)*2);
	customerTable->attrSize = (int *) malloc(sizeof(int)*2);
	customerTable->content = (char **) malloc(sizeof(char *)*2);
	customerTable->attrSize[0] = 15;
	customerTable->attrType[0] = STRING;
	outFd = open("CUSTOMER4",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	customerTable->tupleNum = outSize/15;
	customerTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(customerTable->content[0],outTable,outSize);
	close(outFd);
	customerTable->attrSize[1] = sizeof(int);
	customerTable->attrType[1] = INT;
	outFd = open("CUSTOMER0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	customerTable->content[1] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(customerTable->content[1],outTable,outSize);
	close(outFd);
	customerTable->tupleSize = 0 + 15 + sizeof(int);
	customerTable->dataPos = MEM;

	struct relation customerRel;
	customerRel.hasWhere = 1;
	customerRel.whereAttrNum = 1;
	customerRel.whereAttrType = (int *)malloc(sizeof(int)*1);
	customerRel.whereAttrSize = (int *)malloc(sizeof(int)*1);
	customerRel.content = (char **)malloc(sizeof(char *)*1);
	customerRel.whereAttrType[0] = STRING;
	customerRel.whereAttrSize[0] = 12;
	outFd = open("CUSTOMER5",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	customerRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(customerRel.content[0],outTable,outSize);
	close(outFd);
	customerRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(customerRel.filter)->nested = 0;
	(customerRel.filter)->andOr = EXP;
	(customerRel.filter)->exp[0].index = 0;
	(customerRel.filter)->exp[0].relation = EQ;
	strcpy((customerRel.filter)->exp[0].content,"AMERICA");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan(customerTable, &customerRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);
	totalTime += (end.tv_sec - start.tv_sec) * BILLION + end.tv_nsec - start.tv_nsec;

	partTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	partTable->totalAttr = 1;
	partTable->attrType = (int *) malloc(sizeof(int)*1);
	partTable->attrSize = (int *) malloc(sizeof(int)*1);
	partTable->content = (char **) malloc(sizeof(char *)*1);
	partTable->attrSize[0] = sizeof(int);
	partTable->attrType[0] = INT;
	outFd = open("PART0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	partTable->tupleNum = outSize/sizeof(int);
	partTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(partTable->content[0],outTable,outSize);
	close(outFd);
	partTable->tupleSize = 0 + sizeof(int);
	partTable->dataPos = MEM;

	struct relation partRel;
	partRel.hasWhere = 1;
	partRel.whereAttrNum = 1;
	partRel.whereAttrType = (int *)malloc(sizeof(int)*2);
	partRel.whereAttrSize = (int *)malloc(sizeof(int)*2);
	partRel.content = (char **)malloc(sizeof(char *)*2);
	partRel.whereAttrType[0] = STRING;
	partRel.whereAttrSize[0] = 6;
	outFd = open("PART2",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	partRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(partRel.content[0],outTable,outSize);
	close(outFd);
	partRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(partRel.filter)->nested = 0;
	(partRel.filter)->andOr = OR;
	(partRel.filter)->exp[0].index = 0;
	(partRel.filter)->exp[0].relation = EQ;
	strcpy((partRel.filter)->exp[0].content,"MFGR#1");

	(partRel.filter)->exp[1].index = 0;
	(partRel.filter)->exp[1].relation = EQ;
	strcpy((partRel.filter)->exp[1].content,"MFGR#2");

	clock_gettime(CLOCK_REALTIME,&start);
	tableScan(partTable, &partRel, &pp);
	clock_gettime(CLOCK_REALTIME,&end);
	totalTime += (end.tv_sec - start.tv_sec) * BILLION + end.tv_nsec - start.tv_nsec;

	ddateTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	ddateTable->totalAttr = 2;
	ddateTable->attrType = (int *) malloc(sizeof(int)*2);
	ddateTable->attrSize = (int *) malloc(sizeof(int)*2);
	ddateTable->content = (char **) malloc(sizeof(char *)*2);
	ddateTable->attrSize[0] = sizeof(int);
	ddateTable->attrType[0] = INT;
	outFd = open("DDATE4",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	ddateTable->tupleNum = outSize/sizeof(int);
	ddateTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(ddateTable->content[0],outTable,outSize);
	close(outFd);
	ddateTable->attrSize[1] = sizeof(int);
	ddateTable->attrType[1] = INT;
	outFd = open("DDATE0",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	ddateTable->content[1] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize,PROT_READ,MAP_SHARED,outFd,0);
	memcpy(ddateTable->content[1],outTable,outSize);
	close(outFd);
	ddateTable->tupleSize = 0 + sizeof(int) + sizeof(int);
	ddateTable->dataPos = MEM;

	struct tableNode *lineorderTable = (struct tableNode*)malloc(sizeof(struct tableNode));
	lineorderTable->totalAttr = 6;
	lineorderTable->attrType = (int *) malloc(sizeof(int)*6);
	lineorderTable->attrSize = (int *) malloc(sizeof(int)*6);
	lineorderTable->content = (char **) malloc(sizeof(char *)*6);
	outFd = open("LINEORDER12",O_RDONLY);
	outSize = lseek(outFd,0,SEEK_END);
	lineorderTable->tupleNum = outSize /sizeof(int);
	lineorderTable->attrType[0] = INT;
	lineorderTable->attrSize[0] = sizeof(int);
	lineorderTable->attrType[1] = INT;
	lineorderTable->attrSize[1] = sizeof(int);
	lineorderTable->attrType[2] = INT;
	lineorderTable->attrSize[2] = sizeof(int);
	lineorderTable->attrType[3] = INT;
	lineorderTable->attrSize[3] = sizeof(int);
	lineorderTable->attrType[4] = INT;
	lineorderTable->attrSize[4] = sizeof(int);
	lineorderTable->attrType[5] = INT;
	lineorderTable->attrSize[5] = sizeof(int);
	lineorderTable->tupleSize = 0 + sizeof(int) + sizeof(int) + sizeof(int) + sizeof(int) + sizeof(int) + sizeof(int);
	int pass = lineorderTable->tupleNum / BLOCK + 1;
	long tupleUnit = lineorderTable->tupleNum / pass;
	long nextScan = tupleUnit;
	long restTuple = lineorderTable->tupleNum;
	long tupleOffset = 0;
	long offset = 0;

	for(int i=0;i<pass;i++){

		if(restTuple < nextScan)
			nextScan = restTuple;

		lineorderTable->tupleNum = nextScan;
		outFd = open("LINEORDER12", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[0] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[0], outTable+offset, outSize);
		close(outFd);

		outFd = open("LINEORDER13", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[1] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[1], outTable+offset, outSize);
		close(outFd);

		outFd = open("LINEORDER5", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[2] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[2], outTable+offset, outSize);
		close(outFd);

		outFd = open("LINEORDER3", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[3] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[3], outTable+offset, outSize);
		close(outFd);

		outFd = open("LINEORDER2", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[4] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[4], outTable+offset, outSize);
		close(outFd);

		outFd = open("LINEORDER4", O_RDONLY);
		offset = tupleOffset*sizeof(int);
		outSize = nextScan*sizeof(int);
		outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
		lineorderTable->content[5] = (char *) malloc(outSize);
		memcpy(lineorderTable->content[5], outTable+offset, outSize);
		close(outFd);

		clock_gettime(CLOCK_REALTIME,&start);
		struct joinNode jNode0;
		jNode0.leftTable = lineorderTable;
		jNode0.rightTable = supplierTable;
		jNode0.rightOutputAttrNum = 0;
		jNode0.leftOutputAttrNum = 5;
		jNode0.leftOutputAttrType = (int *)malloc(sizeof(int)*5);
		jNode0.leftOutputIndex = (int *)malloc(sizeof(int)*5);
		jNode0.tupleSize = 0;
		jNode0.leftOutputIndex[0] = 0;
		jNode0.leftOutputAttrType[0] = INT;
		jNode0.tupleSize += lineorderTable->attrSize[0];
		jNode0.leftOutputIndex[1] = 1;
		jNode0.leftOutputAttrType[1] = INT;
		jNode0.tupleSize += lineorderTable->attrSize[1];
		jNode0.leftOutputIndex[2] = 2;
		jNode0.leftOutputAttrType[2] = INT;
		jNode0.tupleSize += lineorderTable->attrSize[2];
		jNode0.leftOutputIndex[3] = 3;
		jNode0.leftOutputAttrType[3] = INT;
		jNode0.tupleSize += lineorderTable->attrSize[3];
		jNode0.leftOutputIndex[4] = 4;
		jNode0.leftOutputAttrType[4] = INT;
		jNode0.tupleSize += lineorderTable->attrSize[4];
		jNode0.rightOutputAttrType = (int *)malloc(sizeof(int)*0);
		jNode0.rightOutputIndex = (int *)malloc(sizeof(int)*0);
		jNode0.rightKeyIndex = 0;
		jNode0.leftKeyIndex = 5;
		struct tableNode *join0 = hashJoin(&jNode0, &pp);

		struct joinNode jNode1;
		jNode1.leftTable = join0;
		jNode1.rightTable = customerTable;
		jNode1.rightOutputAttrNum = 1;
		jNode1.leftOutputAttrNum = 4;
		jNode1.leftOutputAttrType = (int *)malloc(sizeof(int)*4);
		jNode1.leftOutputIndex = (int *)malloc(sizeof(int)*4);
		jNode1.tupleSize = 0;
		jNode1.leftOutputIndex[0] = 0;
		jNode1.leftOutputAttrType[0] = INT;
		jNode1.tupleSize += join0->attrSize[0];
		jNode1.leftOutputIndex[1] = 1;
		jNode1.leftOutputAttrType[1] = INT;
		jNode1.tupleSize += join0->attrSize[1];
		jNode1.leftOutputIndex[2] = 2;
		jNode1.leftOutputAttrType[2] = INT;
		jNode1.tupleSize += join0->attrSize[2];
		jNode1.leftOutputIndex[3] = 3;
		jNode1.leftOutputAttrType[3] = INT;
		jNode1.tupleSize += join0->attrSize[3];
		jNode1.rightOutputAttrType = (int *)malloc(sizeof(int)*1);
		jNode1.rightOutputIndex = (int *)malloc(sizeof(int)*1);
		jNode1.rightOutputIndex[0] = 0;
		jNode1.rightOutputAttrType[0] = STRING;
		jNode1.tupleSize += customerTable->attrSize[0];
		jNode1.rightKeyIndex = 0;
		jNode1.leftKeyIndex = 4;
		struct tableNode *join1 = hashJoin(&jNode1, &pp);

		struct joinNode jNode2;
		jNode2.leftTable = join1;
		jNode2.rightTable = partTable;
		jNode2.rightOutputAttrNum = 0;
		jNode2.leftOutputAttrNum = 4;
		jNode2.leftOutputAttrType = (int *)malloc(sizeof(int)*4);
		jNode2.leftOutputIndex = (int *)malloc(sizeof(int)*4);
		jNode2.tupleSize = 0;
		jNode2.leftOutputIndex[0] = 0;
		jNode2.leftOutputAttrType[0] = STRING;
		jNode2.tupleSize += join1->attrSize[0];
		jNode2.leftOutputIndex[1] = 1;
		jNode2.leftOutputAttrType[1] = INT;
		jNode2.tupleSize += join1->attrSize[1];
		jNode2.leftOutputIndex[2] = 2;
		jNode2.leftOutputAttrType[2] = INT;
		jNode2.tupleSize += join1->attrSize[2];
		jNode2.leftOutputIndex[3] = 3;
		jNode2.leftOutputAttrType[3] = INT;
		jNode2.tupleSize += join1->attrSize[3];
		jNode2.rightOutputAttrType = (int *)malloc(sizeof(int)*0);
		jNode2.rightOutputIndex = (int *)malloc(sizeof(int)*0);
		jNode2.rightKeyIndex = 0;
		jNode2.leftKeyIndex = 4;
		struct tableNode *join2 = hashJoin(&jNode2, &pp);

		struct joinNode jNode3;
		jNode3.leftTable = join2;
		jNode3.rightTable = ddateTable;
		jNode3.rightOutputAttrNum = 1;
		jNode3.leftOutputAttrNum = 3;
		jNode3.leftOutputAttrType = (int *)malloc(sizeof(int)*3);
		jNode3.leftOutputIndex = (int *)malloc(sizeof(int)*3);
		jNode3.tupleSize = 0;
		jNode3.leftOutputIndex[0] = 0;
		jNode3.leftOutputAttrType[0] = STRING;
		jNode3.tupleSize += join2->attrSize[0];
		jNode3.leftOutputIndex[1] = 1;
		jNode3.leftOutputAttrType[1] = INT;
		jNode3.tupleSize += join2->attrSize[1];
		jNode3.leftOutputIndex[2] = 2;
		jNode3.leftOutputAttrType[2] = INT;
		jNode3.tupleSize += join2->attrSize[2];
		jNode3.rightOutputAttrType = (int *)malloc(sizeof(int)*1);
		jNode3.rightOutputIndex = (int *)malloc(sizeof(int)*1);
		jNode3.rightOutputIndex[0] = 0;
		jNode3.rightOutputAttrType[0] = INT;
		jNode3.tupleSize += ddateTable->attrSize[0];
		jNode3.rightKeyIndex = 0;
		jNode3.leftKeyIndex = 3;
		struct tableNode *join3 = hashJoin(&jNode3, &pp);

		freeTable(jNode0);
		freeTable(jNode1);
		freeTable(jNode2);
		freeTable(jNode3);
		free(join3->content[0]);
		free(join3->content[1]);
		free(join3->content[2]);
		free(join3->content[3]);
		free(join3->content[4]);
		free(join3->content[5]);
		tupleOffset += tupleUnit;
		restTuple -=nextScan;
		clock_gettime(CLOCK_REALTIME,&end);
	}

}

